package org.niit.flume;

/*

Flume  自定拦截器，根据发送数据的内容不同添加事件中的Header,根据Header中的Key决定发往哪个Flume
 */

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class TypeInterceptor implements Interceptor {

    //声明存放事件的集合
    private List<Event> addHeaderEvents;

    //对拦截器进行初始化
    @Override
    public void initialize() {
        addHeaderEvents = new ArrayList<>();
    }

    //拦截单个事件方法
    @Override
    public Event intercept(Event event) {
        //1.获得事件中的 body信息
        String body = new String(event.getBody());
        //2.获得事件中的 header 信息
        Map<String, String> headers = event.getHeaders();

        //3.根据body中是否含有MYFLUME来决定添加怎样的头信息
        if(body.contains("MYFLUME")){
            //4.添加头信息
            headers.put("type","first");
        }else{
            headers.put("type","second");
        }
        return event;
    }

    //拦截多个事件的方法
    @Override
    public List<Event> intercept(List<Event> list) {
        //1.清空集合
        addHeaderEvents.clear();
        //2.遍历 list
        for (Event event:list) {
            Event newEvent = intercept(event);
            addHeaderEvents.add(newEvent);
        }
        return addHeaderEvents;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder{


        @Override
        public Interceptor build() {
            return new TypeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

}


